#include "ChunkStore.h"
#include "Service.h"

#include "Cycles.h"

int 
ChunkServer::ConnectToMaster()
{
	master_sess_id_ = rpc_client_->ConnectSess(master_uri_, NULL, NULL);
}

static void 
_heartbeat_cb(void* self, void* ex, void* none)
{
	RdmaRpc* rpc = static_cast<RdmaRpc*>(self);
	rpc->FreeMsg(static_cast<RpcExecution*>(ex)->req_msg);
}

static int 
_heartbeat_poller_fn(void* ctx)
{
	//TODO:
	//never send hearbeat
	ChunkServer *cs = static_cast<ChunkServer*>(ctx);
	if(!cs->last()){
		cs->set_last(Cycles::rdtsc());
	}
	
	if(Cycles::to_seconds(Cycles::rdtsc() - cs->last(), 0) > cs->heartbeat_interval() - 1){
		RpcMsg *rpc_header = static_cast<RpcMsg*>(cs->rpc_client()->AllocMsg());
		cs->prepare_heartbeat_msg(rpc_header);
		cs->rpc_client()->RpcCliWriteReq(cs->master_sess_id(), rpc_header, _heartbeat_cb, NULL);
		cs->set_last(Cycles::rdtsc());
	}
		
	
	return -1;
}

void
ChunkServer::RegisterHeartbeatPoller()
{
	heartbeat_poller_ = spdk_poller_register(_heartbeat_poller_fn, this, 1000000);
}

static void 
_cs_unload_cb(void* cb_arg, int bserrno)
{
	ChunkServer *cs = static_cast<ChunkServer*>(cb_arg);
	if (bserrno) {
		SPDK_ERRLOG("Error %d unloading the blobstore\n", bserrno);
	}
	SPDK_NOTICELOG("mem : %x\n", cs->mem());
	spdk_dma_free(cs->mem());
	if (cs->rc() < 0) {
		spdk_app_stop(-1);
	}
}

static void 
_cs_cleanup(ChunkServer* cs)
{
	SPDK_NOTICELOG("chunkserver cleanup\n");
	cs->FreeConf();
	if (cs->thread()) {
		cs->UnRegisterRpcConnPoller();
		cs->UnRegisterBlobstoreIoPoller();
		cs->UnRegisterRpcIoPoller();
		//spdk_poller_unregister(&cs.heartbeat_poller);
	}
	if (cs->rpc_server()) {
		cs->rpc_server()->Destroy();
		delete cs->rpc_server();
	}
	if (cs->rpc_client()) {
		cs->rpc_client()->Destroy();
		delete cs->rpc_client();
	}

	if(cs->mm())
	{
		cs->mm()->Destroy();
		delete cs->mm();
	}

	if (cs->io_channel()) {
		spdk_bs_free_io_channel(cs->io_channel());
	}
	if (cs->bs()) {
		cs->set_rc(-1);
		
		spdk_bs_unload(cs->bs(), _cs_unload_cb, cs);
	}
	else {
		spdk_app_stop(-1);

	}
}

int 
ChunkServer::ParseConf(const char* conf)
{
	FILE* fp;
	char opt_name[100];
	char opt_val[100];
	if ((fp = fopen(conf, "r")) != NULL) {
		while (fscanf(fp, "%s\t=\t%s", opt_name, opt_val) != EOF) {
			//TODO:
			SPDK_NOTICELOG("CONFIG : %s = %s\n", opt_name, opt_val);
			if (!strcmp(opt_name, "mem_size")) {
				mem_size_ = atoi(opt_val);
			}
			else if (!strcmp(opt_name, "chunk_size")) {
				chunk_size_ = atoi(opt_val);
			}
			else if (!strcmp(opt_name, "heartbeat_interval")) {
				heartbeat_interval_ = atoi(opt_val);
			}
			else if (!strcmp(opt_name, "cs_num")) {
				cs_num_ = atoi(opt_val);
			}
			else if (!strcmp(opt_name, "bdev_name")) {
				strcpy(bdev_name_, opt_val);
			}
			else if (!strcmp(opt_name, "cs_id2spdk_uri")) {
				char id[5];
				char ip[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id, ip);
				SPDK_NOTICELOG("id:%s, ip:%s\n",id,ip);
				spdk_rpc_uri_ = strdup(ip);
			}
			else if (!strcmp(opt_name, "cs_id2uri")) {
				char id[5];
				char ip[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id, ip);
				SPDK_NOTICELOG("id:%s, ip:%s\n",id,ip);
				cs_uri_[atoi(id)] = strdup(ip);
			}
			else if (!strcmp(opt_name, "master_id2uri")) {
				char id[5];
				char ip[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id, ip);
				SPDK_NOTICELOG("id:%s, ip:%s\n",id,ip);
				master_uri_ = strdup(ip);
			}
			
			else if (!strcmp(opt_name, "cs_bdev_json")) {
				bdev_json_file_ = strdup(opt_val);
			}

			else {
				SPDK_ERRLOG("Unrecognized config option : %s\n", opt_name);
				return 0;
			}

			
		}
		fclose(fp);
		return 1;
	}
	else {
		SPDK_ERRLOG("Failed to open conf file : %s", conf);
		return 0;
	}
}

static int 
_rpc_conn_poller_fn(void* ctx)
{
	return static_cast<RdmaRpc*>(ctx)->AcceptSess();
}

void
ChunkServer::RegisterRpcConnPoller()
{
	rpc_conn_poller_ = spdk_poller_register(_rpc_conn_poller_fn, rpc_server_, 1000000);
}

static int 
_rpc_io_poller_fn(void* ctx)
{
	ChunkServer *cs = static_cast<ChunkServer*>(ctx);
	cs->rpc_server()->RpcServerHandleRx();
	cs->rpc_server()->RpcServerHandleTx();
	cs->rpc_client()->RpcCliHandleRx();
	cs->rpc_client()->RpcCliHandleTx();
	return -1;
}

void
ChunkServer::RegisterRpcIoPoller()
{
	rpc_io_poller_ = spdk_poller_register(_rpc_io_poller_fn, this, 0);
}

static int 
_bs_io_poller_fn(void* ctx)
{
	//TODO:
	return -1;
}

void
ChunkServer::RegisterBlobstoreIoPoller()
{
	blobstore_io_poller_ = spdk_poller_register(_bs_io_poller_fn, NULL, 0);
}

static void
_get_next_blob(void* cb_arg, struct spdk_blob* blob, int bserrno);

static void 
_open_blob_cb(void* cb_arg, struct spdk_blob* blob, int bserrno)
{
	ChunkServer *cs = static_cast<ChunkServer*>(cb_arg);
	SPDK_NOTICELOG("Open %lu successfully\n", spdk_blob_get_id(blob));
	cs->id2chunk()[spdk_blob_get_id(blob)] = blob;
	cs->blobs_.emplace_back(spdk_blob_get_id(blob));

	spdk_bs_iter_next(cs->bs(), blob, _get_next_blob, cs);
}

static void
_get_next_blob(void* cb_arg, struct spdk_blob* blob, int bserrno)
{
    ChunkServer *cs = static_cast<ChunkServer*>(cb_arg);
	if(bserrno)
	{
		SPDK_NOTICELOG("Failed to get next blob, maybe is end, total open %lu blobs\n", cs->id2chunk().size());
	}
	else
	{
		spdk_bs_open_blob(cs->bs(), spdk_blob_get_id(blob), _open_blob_cb, cs);
	}

}




static void
_cs_init_cb(void* cb_arg, struct spdk_blob_store* bs, int bserrno)
{
	ChunkServer *cs = static_cast<ChunkServer*>(cb_arg);
	cs->set_bs(bs);
	//create io channel to submit io
	
	cs->set_io_channel(spdk_bs_alloc_io_channel(bs));
	if (!cs->io_channel()) {
		_cs_cleanup(cs);
		SPDK_ERRLOG("Failed to create io channel\n");
		return;
	}

	cs->parse_bs_inf();


	//create pending buffer and id2chunk
	cs->id2chunk().clear();

	//init MiniRPC
	MemoryManager *mm = new MemoryManager(cs->mem(), cs->mem_size(), 256 << 10);
	cs->set_mm(mm);
	if(mm->Init() < 0)
    {
        SPDK_ERRLOG("Failed to init MemoryManager...\n");
		_cs_cleanup(cs);
        return ;
    }
	cs->set_rpc_server(new RdmaRpc(*mm));
    
	if (!cs->rpc_server()) 
	{
		SPDK_ERRLOG("Failed to create rpc server\n");
		_cs_cleanup(cs);
		return;
	}
	cs->rpc_server()->Init(RPC_SERVER | RPC_RDMA);
    cs->rpc_server()->BindSess(cs->cs_uri()[cs->cs_id()]);

	cs->set_rpc_client(new RdmaRpc(*mm));

	if (!cs->rpc_client()) {
		SPDK_ERRLOG("Failed to create rpc client\n");
		_cs_cleanup(cs);
		return;
	}
    cs->rpc_client()->Init(RPC_RDMA);

	//cs.master_sess_id = rpc_create_session(cs.rpc_client, cs.master_uri, master_session_cb, NULL);
	

	// for(int i = 0;cs_uri[i]!=NULL;i++){
	//		if(i!=cs_id){
	//			cs.cs_sess_id[i] = rpc_create_session(cs.minirpc,cs.cs_uri[i]);
	//		}
	//}
	cs->rpc_server()->RpcServiceRegister(S_CHUNK_IO_SERVICE, cs_object_io_service);
	cs->rpc_server()->RpcServiceProviderRegister(S_CHUNK_IO_SERVICE, cs);
	//cs->rpc_server()->RpcServiceRegister(S_CLIENT_AUTH_SERVICE, cs_client_auth_service);
	//cs->rpc_server()->RpcServiceProviderRegister(S_CLIENT_AUTH_SERVICE, cs_client_auth_service);
    cs->ConnectToMaster();

	//register poller in spdk reactor
	cs->set_thread(spdk_get_thread());
	cs->RegisterRpcConnPoller();
	cs->RegisterRpcIoPoller();
	cs->RegisterBlobstoreIoPoller();
	cs->RegisterHeartbeatPoller();

	//cs.heartbeat_poller = spdk_poller_register(heartbeat_poller_fn, cs.rpc_client, 1000000);
	if(cs->is_restart() == true)
	{
	    spdk_bs_iter_first(bs, _get_next_blob, cs);
	}
}

static void
_close_blob_cb(void *arg, int bserrno)
{
	ChunkServer *cs = static_cast<ChunkServer*>(arg);
    
    SPDK_NOTICELOG("Close blob %d successfully\n", cs->close_blob_num());
    cs->set_close_blob_num(cs->close_blob_num()-1);
	
    if(0 == cs->close_blob_num())
    { 
        SPDK_NOTICELOG("Close all blob, unload bs\n");
		if (cs->io_channel()) {
			spdk_bs_free_io_channel(cs->io_channel());
		}
		if (cs->bs()) {
			int rc = 1;
			spdk_bs_unload(cs->bs(), _cs_unload_cb, arg);
		}

    }
}


static void
_base_bdev_event_cb(enum spdk_bdev_event_type type, struct spdk_bdev* bdev,
	void* event_ctx)
{
	ChunkServer *cs = static_cast<ChunkServer*>(event_ctx);
	if (type == SPDK_BDEV_EVENT_REMOVE) {
		SPDK_WARNLOG("Remove bdev and then unload blobstore\n");
		cs->FreeConf();
		if (cs->thread()) {
			cs->UnRegisterRpcConnPoller();
			cs->UnRegisterBlobstoreIoPoller();
			cs->UnRegisterRpcIoPoller();
			//spdk_poller_unregister(&cs.heartbeat_poller);
		}
		if (cs->rpc_server()) {
			cs->rpc_server()->Destroy();
			delete cs->rpc_server();
		}
		if (cs->rpc_client()) {
			cs->rpc_client()->Destroy();
			delete cs->rpc_client();
		}
		if(cs->mm())
		{
			cs->mm()->Destroy();
			delete cs->mm();
		}
		const std::unordered_map<uint64_t, struct spdk_blob*> &map_ref = cs->id2chunk();
        if(map_ref.size())
        { 
            cs->set_close_blob_num(map_ref.size());
            for(auto iter = map_ref.cbegin(); iter != map_ref.end(); ++iter)
            {
                struct spdk_blob *close_blob = iter->second;
                spdk_blob_close(close_blob, _close_blob_cb, cs);
            }
        }
        else{
            if (cs->io_channel()) {
                spdk_bs_free_io_channel(cs->io_channel());
            }
            if (cs->bs()) {
                cs->set_rc(1);
                spdk_bs_unload(cs->bs(), _cs_unload_cb, cs);

            }
        }
		
		
	}
	else {
		SPDK_WARNLOG("Unsupported bdev event: type %d\n", type);
	}

}


static void 
_cs_init(void* arg1)
{
	ChunkServer *cs = static_cast<ChunkServer*>(arg1);
	int rc;
	int cs_id = cs->cs_id();
	//spdk alloc hugepage
	cs->set_mem(spdk_dma_zmalloc(cs->mem_size(), 0x1 << 12, NULL));
	if (!cs->mem()) {
		SPDK_ERRLOG("Failed to allocate buffer\n");
		_cs_cleanup(cs);
		return;
	}

	//init blobstore
	struct spdk_bs_dev* bs_dev = NULL;
	struct spdk_bdev* mybdev = NULL;
	for(mybdev = spdk_bdev_first(); mybdev != NULL; mybdev = spdk_bdev_next(mybdev)){
		if (!strcmp(cs->bdev_name(), spdk_bdev_get_name(mybdev))) {
			SPDK_NOTICELOG("blob bdev %s has been found\n", cs->bdev_name());
			break;
		}
	}
	if (mybdev == NULL) {
		SPDK_ERRLOG("blob bdev %s not exist!!", cs->bdev_name());
		_cs_cleanup(cs);
		return;
	}
	SPDK_NOTICELOG("bdev->required_alignment: %d\n", mybdev->required_alignment);
	
	rc = spdk_bdev_create_bs_dev_ext(cs->bdev_name(), _base_bdev_event_cb, cs, &bs_dev);
	if (rc != 0) {
		SPDK_ERRLOG("Could not create blob bdev, %s!!\n",
			spdk_strerror(-rc));
		_cs_cleanup(cs);
		return;
	}
	cs->set_bs_dev(bs_dev);
	SPDK_NOTICELOG("Attach bdev: %s\n", cs->bdev_name());
	if(cs->is_restart()){
		spdk_bs_load(bs_dev, NULL, _cs_init_cb, cs);

	}
	else
	{
		spdk_bs_init(bs_dev, NULL, _cs_init_cb, cs);
	}
}

int 
main(int argc, char** argv) {
	int rc = 0;
	char ch;
	struct spdk_app_opts opts = {};
    ChunkServer cs;
	while ((ch = getopt(argc, argv, "i:c:rh")) != -1) {
		switch (ch) {
		case 'i':
			cs.set_cs_id(atoi(optarg));
			break;
		case 'c':
			cs.set_cs_conf(strdup(optarg));
			break;
		case 'r':
		    cs.set_is_restart(true);
			break;
		case 'h':
			SPDK_NOTICELOG("\n-i ChunkServer ID\n-c ChunkServer Configure File Name\n-h Help\n");
			return rc;
		default:
			SPDK_ERRLOG("NOT recognized option: %c", ch);
			return rc;
		}
	}
	try
	{
		/* code */
		if(!cs.ParseConf(cs.cs_conf()))
		{
			throw "Failed to parse chunkserver configuration...";
		}
		
	}
	catch(const std::exception& e)
	{
		std::cerr << e.what() << '\n';
		exit(-1);
	}

	spdk_app_opts_init(&opts, sizeof(opts));

	opts.name = "ChunkServer";
	opts.json_config_file = cs.bdev_json_file();
	opts.rpc_addr = cs.spdk_rpc_uri();

	rc = spdk_app_start(&opts, _cs_init, &cs);
	if (rc) {
		SPDK_NOTICELOG("ERROR!\n");
	}
	else {
		SPDK_NOTICELOG("SUCCESS!\n");
	}

	spdk_app_fini();
	return rc;
}